package com.niit.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}



object Spark_Stream_Kafka {
  //作为Kafka的消费者
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkKafka")
    val ssc = new StreamingContext(sparkConf,Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    //定义Kafka参数
    val kafkaPara = Map[String,Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "SP_KF",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG ->  classOf[StringDeserializer].getName,
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer].getName
    )
    //读取Kafka数据创建 DStream
    val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,//配置本地策略
      ConsumerStrategies.Subscribe[String, String](Set("BD2"), kafkaPara)
    )
    //将每条数据的值打印出来
    kafkaDataDS.map(_.value()).print()

    ssc.start()
    ssc.awaitTermination();

  }

}
